package com.zzx.graduation.utils;



import com.zzx.graduation.config.MQDict;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.slf4j.LoggerFactory;


import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Properties;

/**
 * 模板类，具体消息处理交由具体实现类去实现。
 * 可以采用匿名的方式实现。
 */
public abstract class KafkaBatchConsumer {

    private volatile KafkaConsumer<String, String> consumer;
    public volatile boolean flag = true;

    public KafkaBatchConsumer(String topic, String groupId){
        consumer = initConfigAndKafkaConsumer(topic, groupId);
    }
    /**
     * 初始化配置
     */
    private synchronized   KafkaConsumer<String, String> initConfigAndKafkaConsumer(String topic, String groupId) {
        Properties props = new Properties();
        props.put("bootstrap.servers", MQDict.MQ_ADDRESS_COLLECTION);
        props.put("group.id", groupId);
        props.put("enable.auto.commit", MQDict.CONSUMER_ENABLE_AUTO_COMMIT);
        props.put("auto.commit.interval.ms", MQDict.CONSUMER_AUTO_COMMIT_INTERVAL_MS);
        props.put("session.timeout.ms", MQDict.CONSUMER_SESSION_TIMEOUT_MS);
        props.put("max.poll.records", MQDict.CONSUMER_MAX_POLL_RECORDS);
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());

        consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList(topic));
        return consumer;
    }

    public abstract void hanldeRecord(ConsumerRecord<String, String> record);

    public  void receivebatch() {
        //执行5秒
        long now = new Date().getTime();
        while (flag) {
            System.out.println("进入消费while poll 循环中--->>>>");
                ConsumerRecords<String, String> records = consumer.poll(MQDict.CONSUMER_POLL_TIME_OUT);
                records.forEach((ConsumerRecord<String, String> record) -> {
                    System.out.println("消费者收到消息: key ===" + record.key()
                            + " value ====" + record.value()
                            + " topic ===" + record.topic());
                    hanldeRecord(record);
                    try {
                        System.out.println("线程开始休眠-----》");
                        Thread.sleep(100);
                        System.out.println("线程休眠结束------》");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
       }
        System.out.println("线程结束--->>>>");
    }

}
